package com.suxin.kafka.config;

import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;

/**
 * @Classname KafKaConfig
 * @Description [ KafKaConfig ]
 * @Author Tang
 * @Date 2020/11/14 16:19
 * @Created by ASUS
 */
@EnableKafka
@Configuration
public class KafKaOneConfig {

    // bean name 必须是 kafkaListenerContainerFactory 给自动配置类使用

    @Bean
    ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaConsumerFactory());
        return factory;
    }

    /**
     * 批处理的
     * @return
     *
     * 配置参数：
     * max.poll.records: 50  最大处理的记录条数
     * max.poll.interval.ms: 500 每隔多次时间取一次数据，最多只能取50条
     *
     */
    @Bean
    ConcurrentKafkaListenerContainerFactory<String, String> batchKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaConsumerFactory());
        // 批处理
        factory.setBatchListener(true);
        return factory;
    }


    /**
     * 并发处理的
     * @return
     *
     * 配置参数：
     * properties:
     *   partition.assignment.strategy: org.apache.kafka.clients.consumer.RoundRobinAssignor # 并行消费设置这个分区策略为轮询，可以执行最大程度的消费
     *
     */
//    @Bean
    ConcurrentKafkaListenerContainerFactory<String, String> concurrentListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaConsumerFactory());
        // 并发消费，会创建多个MessageContainerListener实例
        factory.setConcurrency(15);
        return factory;
    }


    // bean name 必须是 kafkaConsumerFactory 给自动配置类使用

    @Bean
    @Primary
    public ConsumerFactory<String, String> kafkaConsumerFactory() {
        return new DefaultKafkaConsumerFactory<>(
                consumerOneConfigs().buildConsumerProperties()
        );
    }

    @Bean
    @ConfigurationProperties(prefix = "sys.kafka.one")
    public KafkaProperties consumerOneConfigs() {
        return new KafkaProperties();
    }

}